package databus.client.producer.transaction;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusProducer;
import studio.raptor.databus.client.DatabusProducerBuilder;
import studio.raptor.databus.command.*;

import javax.jms.JMSException;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;

/**
 * Created on 2017/5/25  <br>
 * <p>生产者事务发送</p>
 *
 * @author koalton
 */
public class OrderCenter {
    private static AtomicLong singleProcessSeq = new AtomicLong(System.currentTimeMillis() - 1494085554128L);

    DatabusProducer producer = null;

    public static final SimpleDateFormat Y2MS = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private static Logger log = LoggerFactory.getLogger(DatabusProducer.class);

    public OrderCenter() {
        try {
            DatabusConfig config = new DatabusConfig("Order");
            DatabusProducerBuilder producerBuilder = new DatabusProducerBuilder(config);
            producer = producerBuilder.create();
        } catch (JMSException e1) {
            e1.printStackTrace();
        } catch (IOException e) {
            //加载配置文件异常
            e.printStackTrace();
        }
    }

    /**
     * 发送事件后提交事件
     */
    @Test
    public void send()  {
        if (null == producer) {
            System.out.println("初始化生产者异常");
        }
        //新建事件，事件序号为100001,事件名称为AddOrder，事件来源为Order
        Event event = OrderCenter.createEvent(10, 100001, "AddOrder", "Order");
        for (int i = 1; i <= 10; i++) {
            try {
                event.getEventMetaData().setBusinessNo(nextSeq()+"");
                producer.sendWithNoCommit(event);
                log.info("send finished .");
                producer.commit();
                log.info("commit finished .");
                LockSupport.parkUntil(System.currentTimeMillis()+7*1000);
            } catch (JMSException e) {
                log.error("发送事件时出现异常：",e);
                try {
                    producer.rollback();
                } catch (JMSException e1) {
                    log.error("回滚时发生异常：",e1);
                }
            }
        }
        System.out.println("等待30秒，此期间恢复连接，看是否能够继续发送消息 =====================");
        LockSupport.parkUntil(System.currentTimeMillis()+30*1000);
        try {
            event.getEventBusinessData().setContent("最后一条，看下是不是自动重连了");
            producer.sendWithNoCommit(event);
            producer.commit();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
    /**
     * 发送事件内容选项
     */
    @Test
    public void sendWithProperty() throws JMSException {
        if (null == producer) {
            System.out.println("初始化生产者异常");
        }
        //新建事件，事件序号为100001,事件名称为AddOrder，事件来源为Order
        Event event = OrderCenter.createEvent(10, 100001, "AddOrder", "Order");
        try {
        for (int i = 0; i < 10; i++) {
            MessagePropertyBuilder builder = new MessagePropertyBuilder();
            builder.addStringProperty("name1","name1");
            builder.addStringProperty("name2","name2");
            builder.addIntProperty("name3",1);
            builder.addLongProperty("name4",4L);
            MessageProperty messageProperty = builder.build();
            event.getEventMetaData().setBusinessNo(nextSeq()+"");
            producer.sendWithNoCommit(event,messageProperty);
            producer.commit();
            System.out.println("Send event and do commit.");
        }
        } catch (JMSException e) {
            producer.rollback();
            e.printStackTrace();
        }

    }

    private static long nextSeq() {
        long nextSeq = singleProcessSeq.getAndIncrement();
        return nextSeq;
    }

    private static Event createEvent(int eventSize, Integer eventKindId, String eventName, String eventSource) {
        //设定基础数据
        //创建基础数据类型
        EventBasicData basicData = new EventBasicData();
        basicData.setName(eventName);
        basicData.setKindId(eventKindId);
        basicData.setSource(eventSource);
        basicData.setDeliverTime("2017-07-27 14:06:22");

        //创建业务数据
        EventBusinessData businessData = new EventBusinessData();
        businessData.setContent(createStringMsg(eventSize));
        businessData.setDataType(EventBusinessDataType.DATATYPE_JSON);
        businessData.setProvideType(EventBusinessDataProvideType.PROVIDE_TYPE_CARRY);
        businessData.setDataService("nothing ");

        //创建事件元数据形式
        EventMetaData metaData = new EventMetaData();
        metaData.setBusinessNo("" + nextSeq());
        metaData.setRegionId(12L);
        metaData.setBusinessType(EventMetaDataBusinessType.BUSINESS_TYPE_NRM);
        Map<String,String> map = new HashMap<String,String>();
        map.put("aa","aa");
        metaData.setPublicMap(map);
        metaData.setOfferInstId(new Long[]{1L,3L});
        metaData.setOfferSpecAttrs("傻逼格式");


        //可以定义为 EventHelper类
        Event event = new Event(basicData,businessData,metaData);
        return event;
    }

    /**
     * 生成字符串
     *
     * @param size 字符串大小，单位KB
     * @return 指定长度的字符串
     */
    private static String createStringMsg(int size) {
        String metaData = "abcdefghijklmnopqrstuvwxyz";
        int UNIT = 1024;
//        UNIT = 1 ;
        StringBuilder sbd = new StringBuilder(size * UNIT);
        for (int i = 0; i < size * UNIT; i++) {
            sbd.append(metaData.charAt(new Random().nextInt(26)));
        }
        return sbd.toString();
    }
}
